{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "76c520c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | default_exp _testing.apache_kafka_broker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "38d47b16",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "import asyncio\n",
    "import re\n",
    "import platform\n",
    "import socket\n",
    "import subprocess  # nosec\n",
    "from contextlib import asynccontextmanager\n",
    "from datetime import datetime, timedelta\n",
    "from os import environ\n",
    "from pathlib import Path\n",
    "from tempfile import TemporaryDirectory\n",
    "from typing import *\n",
    "\n",
    "import asyncer\n",
    "import nest_asyncio\n",
    "\n",
    "from fastkafka._components._subprocess import terminate_asyncio_process\n",
    "from fastkafka._components.helpers import in_notebook\n",
    "from fastkafka._components.logger import get_logger\n",
    "from fastkafka._components.meta import delegates, export, filter_using_signature, patch\n",
    "from fastkafka._components.test_dependencies import check_java, check_kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c74ed82b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import shlex\n",
    "from tempfile import TemporaryDirectory\n",
    "\n",
    "import pytest\n",
    "\n",
    "from fastkafka._aiokafka_imports import AIOKafkaConsumer, AIOKafkaProducer\n",
    "from fastkafka._components.helpers import change_dir\n",
    "from fastkafka._components.logger import suppress_timestamps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b81062e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "if in_notebook():\n",
    "    from tqdm.notebook import tqdm\n",
    "else:\n",
    "    from tqdm import tqdm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "49f95ea6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "logger = get_logger(__name__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "687ef020",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] __main__: ok\n"
     ]
    }
   ],
   "source": [
    "suppress_timestamps()\n",
    "logger = get_logger(__name__, level=20)\n",
    "logger.info(\"ok\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5fded319",
   "metadata": {},
   "source": [
    "### Local Kafka"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d552eb74",
   "metadata": {},
   "source": [
    "#### Kafka and zookeeper config helpers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d1fa44d1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "def get_zookeeper_config_string(\n",
    "    data_dir: Union[str, Path],  # the directory where the snapshot is stored.\n",
    "    zookeeper_port: int = 2181,  # the port at which the clients will connect\n",
    ") -> str:\n",
    "    \"\"\"Generates a zookeeeper configuration string that can be exported to file\n",
    "    and used to start a zookeeper instance.\n",
    "\n",
    "    Args:\n",
    "        data_dir: Path to the directory where the zookeepeer instance will save data\n",
    "        zookeeper_port: Port for clients (Kafka brokes) to connect\n",
    "    Returns:\n",
    "        Zookeeper configuration string.\n",
    "\n",
    "    \"\"\"\n",
    "    zookeeper_data_dir = str((Path(data_dir) / \"zookeeper\").resolve())\n",
    "    if platform.system() == \"Windows\":\n",
    "        zookeeper_data_dir = zookeeper_data_dir.replace(\"\\\\\", \"/\")\n",
    "    zookeeper_config = f\"\"\"dataDir={zookeeper_data_dir}\n",
    "clientPort={zookeeper_port}\n",
    "maxClientCnxns=0\n",
    "admin.enableServer=false\n",
    "\"\"\"\n",
    "\n",
    "    return zookeeper_config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "34bbd0b5",
   "metadata": {},
   "outputs": [],
   "source": [
    "p = Path(\"..\").resolve()\n",
    "data_dir = str(p).replace(\"\\\\\", \"/\") if platform.system() == \"Windows\" else str(p)\n",
    "assert (\n",
    "    get_zookeeper_config_string(data_dir=\"..\")\n",
    "    == f\"\"\"dataDir={data_dir}/zookeeper\n",
    "clientPort=2181\n",
    "maxClientCnxns=0\n",
    "admin.enableServer=false\n",
    "\"\"\"\n",
    ")\n",
    "\n",
    "assert (\n",
    "    get_zookeeper_config_string(data_dir=\"..\", zookeeper_port=100)\n",
    "    == f\"\"\"dataDir={data_dir}/zookeeper\n",
    "clientPort=100\n",
    "maxClientCnxns=0\n",
    "admin.enableServer=false\n",
    "\"\"\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "42d393bc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "def get_kafka_config_string(\n",
    "    data_dir: Union[str, Path], zookeeper_port: int = 2181, listener_port: int = 9092\n",
    ") -> str:\n",
    "    \"\"\"Generates a kafka broker configuration string that can be exported to file\n",
    "    and used to start a kafka broker instance.\n",
    "\n",
    "    Args:\n",
    "        data_dir: Path to the directory where the kafka broker instance will save data\n",
    "        zookeeper_port: Port on which the zookeeper instance is running\n",
    "        listener_port: Port on which the clients (producers and consumers) can connect\n",
    "    Returns:\n",
    "        Kafka broker configuration string.\n",
    "\n",
    "    \"\"\"\n",
    "    kafka_logs_dir = str((Path(data_dir) / \"kafka_logs\").resolve())\n",
    "    if platform.system() == \"Windows\":\n",
    "        kafka_logs_dir = kafka_logs_dir.replace(\"\\\\\", \"/\")\n",
    "    kafka_config = f\"\"\"broker.id=0\n",
    "\n",
    "############################# Socket Server Settings #############################\n",
    "\n",
    "# The address the socket server listens on. If not configured, the host name will be equal to the value of\n",
    "# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092.\n",
    "#   FORMAT:\n",
    "#     listeners = listener_name://host_name:port\n",
    "#   EXAMPLE:\n",
    "#     listeners = PLAINTEXT://your.host.name:9092\n",
    "listeners=PLAINTEXT://:{listener_port}\n",
    "\n",
    "# Listener name, hostname and port the broker will advertise to clients.\n",
    "# If not set, it uses the value for \"listeners\".\n",
    "# advertised.listeners=PLAINTEXT://localhost:{listener_port}\n",
    "\n",
    "# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details\n",
    "#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL\n",
    "\n",
    "# The number of threads that the server uses for receiving requests from the network and sending responses to the network\n",
    "num.network.threads=3\n",
    "\n",
    "# The number of threads that the server uses for processing requests, which may include disk I/O\n",
    "num.io.threads=8\n",
    "\n",
    "# The send buffer (SO_SNDBUF) used by the socket server\n",
    "socket.send.buffer.bytes=102400\n",
    "\n",
    "# The receive buffer (SO_RCVBUF) used by the socket server\n",
    "socket.receive.buffer.bytes=102400\n",
    "\n",
    "# The maximum size of a request that the socket server will accept (protection against OOM)\n",
    "socket.request.max.bytes=104857600\n",
    "\n",
    "\n",
    "############################# Log Basics #############################\n",
    "\n",
    "# A comma separated list of directories under which to store log files\n",
    "log.dirs={kafka_logs_dir}\n",
    "\n",
    "# The default number of log partitions per topic. More partitions allow greater\n",
    "# parallelism for consumption, but this will also result in more files across\n",
    "# the brokers.\n",
    "num.partitions=1\n",
    "\n",
    "# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.\n",
    "# This value is recommended to be increased for installations with data dirs located in RAID array.\n",
    "num.recovery.threads.per.data.dir=1\n",
    "\n",
    "offsets.topic.replication.factor=1\n",
    "transaction.state.log.replication.factor=1\n",
    "transaction.state.log.min.isr=1\n",
    "\n",
    "# The number of messages to accept before forcing a flush of data to disk\n",
    "log.flush.interval.messages=10000\n",
    "\n",
    "# The maximum amount of time a message can sit in a log before we force a flush\n",
    "log.flush.interval.ms=1000\n",
    "\n",
    "# The minimum age of a log file to be eligible for deletion due to age\n",
    "log.retention.hours=168\n",
    "\n",
    "# A size-based retention policy for logs. Segments are pruned from the log unless the remaining\n",
    "# segments drop below log.retention.bytes. Functions independently of log.retention.hours.\n",
    "log.retention.bytes=1073741824\n",
    "\n",
    "# The maximum size of a log segment file. When this size is reached a new log segment will be created.\n",
    "log.segment.bytes=1073741824\n",
    "\n",
    "# The interval at which log segments are checked to see if they can be deleted according to the retention policies\n",
    "log.retention.check.interval.ms=300000\n",
    "\n",
    "# Zookeeper connection string (see zookeeper docs for details).\n",
    "zookeeper.connect=localhost:{zookeeper_port}\n",
    "\n",
    "# Timeout in ms for connecting to zookeeper\n",
    "zookeeper.connection.timeout.ms=18000\n",
    "\n",
    "# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.\n",
    "group.initial.rebalance.delay.ms=0\n",
    "\"\"\"\n",
    "\n",
    "    return kafka_config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2582427a",
   "metadata": {},
   "outputs": [],
   "source": [
    "p = Path(\"..\").resolve()\n",
    "data_dir = str(p).replace(\"\\\\\", \"/\") if platform.system() == \"Windows\" else str(p)\n",
    "actual = get_kafka_config_string(data_dir=\"..\", listener_port=9999)\n",
    "assert f\"log.dirs={data_dir}/kafka_logs\" in actual\n",
    "assert \"listeners=PLAINTEXT://:9999\" in actual"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "873f5b20",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "@export(\"fastkafka.testing\")\n",
    "class ApacheKafkaBroker:\n",
    "    \"\"\"ApacheKafkaBroker class, used for running unique kafka brokers in tests to prevent topic clashing.\"\"\"\n",
    "\n",
    "    @delegates(get_kafka_config_string, but=[\"data_dir\"])\n",
    "    @delegates(get_zookeeper_config_string, keep=True, but=[\"data_dir\"])\n",
    "    def __init__(\n",
    "        self,\n",
    "        topics: Iterable[str] = [],\n",
    "        *,\n",
    "        retries: int = 3,\n",
    "        apply_nest_asyncio: bool = False,\n",
    "        **kwargs: Dict[str, Any],\n",
    "    ):\n",
    "        \"\"\"Initialises the ApacheKafkaBroker object\n",
    "\n",
    "        Args:\n",
    "            topics: List of topics to create after sucessfull Kafka broker startup\n",
    "            retries: Number of retries to create kafka and zookeeper services using random\n",
    "            apply_nest_asyncio: set to True if running in notebook\n",
    "            zookeeper_port: Port for clients (Kafka brokes) to connect\n",
    "            listener_port: Port on which the clients (producers and consumers) can connect\n",
    "        \"\"\"\n",
    "        self.zookeeper_kwargs = filter_using_signature(\n",
    "            get_zookeeper_config_string, **kwargs\n",
    "        )\n",
    "        self.kafka_kwargs = filter_using_signature(get_kafka_config_string, **kwargs)\n",
    "\n",
    "        if \"zookeeper_port\" not in self.zookeeper_kwargs:\n",
    "            self.zookeeper_kwargs[\"zookeeper_port\"] = 2181\n",
    "            self.kafka_kwargs[\"zookeeper_port\"] = 2181\n",
    "\n",
    "        if \"listener_port\" not in self.kafka_kwargs:\n",
    "            self.kafka_kwargs[\"listener_port\"] = 9092\n",
    "\n",
    "        self.retries = retries\n",
    "        self.apply_nest_asyncio = apply_nest_asyncio\n",
    "        self.temporary_directory: Optional[TemporaryDirectory] = None\n",
    "        self.temporary_directory_path: Optional[Path] = None\n",
    "        self.kafka_task: Optional[asyncio.subprocess.Process] = None\n",
    "        self.zookeeper_task: Optional[asyncio.subprocess.Process] = None\n",
    "        self._is_started = False\n",
    "        self.topics: Iterable[str] = topics\n",
    "\n",
    "    @property\n",
    "    def is_started(self) -> bool:\n",
    "        \"\"\"Property indicating whether the ApacheKafkaBroker object is started.\n",
    "\n",
    "        The is_started property indicates if the ApacheKafkaBroker object is currently\n",
    "        in a started state. This implies that Zookeeper and Kafka broker processes have\n",
    "        sucesfully started and are ready for handling events.\n",
    "\n",
    "        Returns:\n",
    "            bool: True if the object is started, False otherwise.\n",
    "        \"\"\"\n",
    "        return self._is_started\n",
    "\n",
    "    @classmethod\n",
    "    def _check_deps(cls) -> None:\n",
    "        \"\"\"Prepares the environment for running Kafka brokers.\n",
    "        Returns:\n",
    "           None\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    async def _start(self) -> str:\n",
    "        \"\"\"Starts a local kafka broker and zookeeper instance asynchronously\n",
    "        Returns:\n",
    "           Kafka broker bootstrap server address in string format: add:port\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    def start(self) -> str:\n",
    "        \"\"\"Starts a local kafka broker and zookeeper instance synchronously\n",
    "        Returns:\n",
    "           Kafka broker bootstrap server address in string format: add:port\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    def stop(self) -> None:\n",
    "        \"\"\"Stops a local kafka broker and zookeeper instance synchronously\"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    async def _stop(self) -> None:\n",
    "        \"\"\"Stops a local kafka broker and zookeeper instance synchronously\n",
    "        Returns:\n",
    "           None\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    def get_service_config_string(self, service: str, *, data_dir: Path) -> str:\n",
    "        \"\"\"Generates a configuration for a service\n",
    "        Args:\n",
    "            data_dir: Path to the directory where the zookeepeer instance will save data\n",
    "            service: \"kafka\" or \"zookeeper\", defines which service to get config string for\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    async def _start_service(self, service: str = \"kafka\") -> None:\n",
    "        \"\"\"Starts the service according to defined service var\n",
    "        Args:\n",
    "            service: \"kafka\" or \"zookeeper\", defines which service to start\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    async def _start_zookeeper(self) -> None:\n",
    "        \"\"\"Start a local zookeeper instance\n",
    "        Returns:\n",
    "           None\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    async def _start_kafka(self) -> None:\n",
    "        \"\"\"Start a local kafka broker\n",
    "        Returns:\n",
    "           None\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    async def _create_topics(self) -> None:\n",
    "        \"\"\"Create missing topics in local Kafka broker\n",
    "        Returns:\n",
    "           None\n",
    "        \"\"\"\n",
    "        raise NotImplementedError\n",
    "\n",
    "    def __enter__(self) -> str:\n",
    "        #         ApacheKafkaBroker._check_deps()\n",
    "        return self.start()\n",
    "\n",
    "    def __exit__(self, *args: Any, **kwargs: Any) -> None:\n",
    "        self.stop()\n",
    "\n",
    "    async def __aenter__(self) -> str:\n",
    "        #         ApacheKafkaBroker._check_deps()\n",
    "        return await self._start()\n",
    "\n",
    "    async def __aexit__(self, *args: Any, **kwargs: Any) -> None:\n",
    "        await self._stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6cbd5808",
   "metadata": {},
   "outputs": [],
   "source": [
    "# print(combine_params(combine_params(ApacheKafkaBroker, get_kafka_config_string), get_zookeeper_config_string).__doc__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "167099d2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "@patch(cls_method=True)  # type: ignore\n",
    "def _check_deps(cls: ApacheKafkaBroker) -> None:\n",
    "    \"\"\"Checks the dependencies required to run Apache KafkaBroker.\n",
    "\n",
    "    Raises:\n",
    "        RuntimeError: If JDK installation or Kafka installation is not found.\n",
    "    \"\"\"\n",
    "    if not check_java():\n",
    "        raise RuntimeError(\n",
    "            \"JDK installation not found! Please install JDK manually or run 'fastkafka testing install_deps'.\"\n",
    "        )\n",
    "    if not check_kafka():\n",
    "        raise RuntimeError(\n",
    "            \"Kafka installation not found! Please install Kafka tools manually or run 'fastkafka testing install_deps'.\"\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b4c383b8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "check_kafka()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d4600ad5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n"
     ]
    }
   ],
   "source": [
    "# TODO: test\n",
    "\n",
    "broker = ApacheKafkaBroker()\n",
    "broker._check_deps()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e5d74671",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "async def run_and_match(\n",
    "    *args: str,\n",
    "    capture: str = \"stdout\",\n",
    "    timeout: int = 5,\n",
    "    pattern: str,\n",
    "    num_to_match: int = 1,\n",
    ") -> asyncio.subprocess.Process:\n",
    "    \"\"\"Runs a command asynchronously and matches the output against a pattern.\n",
    "\n",
    "    Args:\n",
    "        *args: Command-line arguments for the subprocess.\n",
    "        capture: Which output to capture (\"stdout\" or \"stderr\").\n",
    "        timeout: Timeout in seconds for reading the output.\n",
    "        pattern: Regular expression pattern to match in the output.\n",
    "        num_to_match: Number of matches to wait for.\n",
    "\n",
    "    Returns:\n",
    "        The subprocess process object.\n",
    "\n",
    "    Raises:\n",
    "        ValueError: If the capture parameter has an unsupported value.\n",
    "        TimeoutError: If the process times out.\n",
    "        RuntimeError: If the process returns a non-zero return code.\n",
    "    \"\"\"\n",
    "    # Create the subprocess; redirect the standard output\n",
    "    # into a pipe.\n",
    "    matched = 0\n",
    "\n",
    "    if platform.system() == \"Windows\":\n",
    "        proc = await asyncio.create_subprocess_shell(\n",
    "            \" \".join(args),\n",
    "            stdout=asyncio.subprocess.PIPE,\n",
    "            stderr=asyncio.subprocess.PIPE,\n",
    "            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,  # type: ignore\n",
    "        )\n",
    "    else:\n",
    "        proc = await asyncio.create_subprocess_exec(\n",
    "            *args,\n",
    "            stdout=asyncio.subprocess.PIPE,\n",
    "            stderr=asyncio.subprocess.PIPE,\n",
    "        )\n",
    "\n",
    "    # Read one line of output.\n",
    "    t = datetime.now()\n",
    "    while datetime.now() - t < timedelta(seconds=timeout):\n",
    "        try:\n",
    "            if capture == \"stdout\":\n",
    "                data = await asyncio.wait_for(proc.stdout.readline(), timeout=1.0)  # type: ignore\n",
    "            elif capture == \"stderr\":\n",
    "                data = await asyncio.wait_for(proc.stderr.readline(), timeout=1.0)  # type: ignore\n",
    "            else:\n",
    "                raise ValueError(\n",
    "                    f\"Unknown capture param value {capture}, supported values are 'stdout', 'stderr'\"\n",
    "                )\n",
    "            ddata = data.decode(\"utf-8\")\n",
    "\n",
    "            if len(re.findall(pattern, ddata)) > 0:\n",
    "                # print(f\"Matched: {ddata}\")\n",
    "                matched += 1\n",
    "                if matched == num_to_match:\n",
    "                    return proc\n",
    "        except asyncio.exceptions.TimeoutError as e:\n",
    "            pass\n",
    "\n",
    "        if proc.returncode is not None:\n",
    "            stdout, stderr = await proc.communicate()\n",
    "            dstdout = stdout.decode(\"utf-8\")\n",
    "            dstderr = stderr.decode(\"utf-8\")\n",
    "            if proc.returncode == 0:\n",
    "                raise TimeoutError(\n",
    "                    f\"stdout={dstdout}, stderr={dstderr}, returncode={proc.returncode}\"\n",
    "                )\n",
    "            else:\n",
    "                raise RuntimeError(\n",
    "                    f\"stdout={dstdout}, stderr={dstderr}, returncode={proc.returncode}\"\n",
    "                )\n",
    "\n",
    "    await terminate_asyncio_process(proc)\n",
    "\n",
    "    raise TimeoutError()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a682b7da",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 46756...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 46756 terminated.\n"
     ]
    }
   ],
   "source": [
    "cmd = \"import datetime; from time import sleep; sleep(3); print('time is:' + str(datetime.datetime.now()))\"\n",
    "if platform.system() == \"Windows\":\n",
    "    cmd = \"\\\"import datetime; from time import sleep; sleep(3); print('time is:' + str(datetime.datetime.now()))\\\"\"\n",
    "\n",
    "with pytest.raises(TimeoutError):\n",
    "    proc = await run_and_match(\"python3\", \"-c\", cmd, pattern=\"time is\", timeout=1)\n",
    "\n",
    "with pytest.raises(RuntimeError):\n",
    "    proc = await run_and_match(\n",
    "        \"python3\", \"-c\", \"should break on this\", pattern=\"time is\", timeout=5\n",
    "    )\n",
    "\n",
    "proc = await run_and_match(\"python3\", \"-c\", cmd, pattern=\"time is\", timeout=10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d27a4fd1",
   "metadata": {},
   "outputs": [],
   "source": [
    "cmd = \"import datetime; from time import sleep; sleep(3); print('time is:' + str(datetime.datetime.now())); print('time is:' + str(datetime.datetime.now()))\"\n",
    "if platform.system() == \"Windows\":\n",
    "    cmd = \"\\\"import datetime; from time import sleep; sleep(3); print('time is:' + str(datetime.datetime.now())); print('time is:' + str(datetime.datetime.now()))\\\"\"\n",
    "\n",
    "with pytest.raises(TimeoutError):\n",
    "    proc = await run_and_match(\n",
    "        \"python3\", \"-c\", cmd, pattern=\"time is\", timeout=5, num_to_match=3\n",
    "    )\n",
    "\n",
    "\n",
    "proc = await run_and_match(\n",
    "    \"python3\", \"-c\", cmd, pattern=\"time is\", timeout=10, num_to_match=2\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "19ef0a81",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "def is_port_in_use(port: Union[int, str]) -> bool:\n",
    "    \"\"\"\n",
    "    Checks if a port is already in use.\n",
    "\n",
    "    Args:\n",
    "        port (Union[int, str]): The port number to check. It can be provided as an integer or a string.\n",
    "\n",
    "    Returns:\n",
    "        bool: True if the port is in use, False otherwise.\n",
    "    \"\"\"\n",
    "    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n",
    "        return s.connect_ex((\"localhost\", int(port))) == 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e84c49c7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Socket is listening on port 9969\n"
     ]
    }
   ],
   "source": [
    "def bind_to_port(port):\n",
    "    try:\n",
    "        # Create a socket object\n",
    "        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "        # Set the socket option to reuse the address\n",
    "        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n",
    "        # Bind the socket to the specified port\n",
    "        sock.bind((\"localhost\", port))\n",
    "        # Listen for incoming connections\n",
    "        sock.listen(1)\n",
    "        print(f\"Socket is listening on port {port}\")\n",
    "        return sock\n",
    "    except Exception as e:\n",
    "        print(f\"Failed to bind to port {port}: {e}\")\n",
    "        return None\n",
    "\n",
    "\n",
    "test_port = 9969\n",
    "assert not is_port_in_use(port=test_port)\n",
    "\n",
    "s = bind_to_port(test_port)\n",
    "assert is_port_in_use(port=test_port)\n",
    "if s:\n",
    "    s.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c7a076ff",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "def get_free_port() -> str:\n",
    "    \"\"\"Gets a port number which is available and free in the system.\n",
    "\n",
    "    Returns:\n",
    "        The free port number as a string.\n",
    "    \"\"\"\n",
    "    s = socket.socket()\n",
    "    s.bind((\"127.0.0.1\", 0))\n",
    "    port = str(s.getsockname()[1])\n",
    "    s.close()\n",
    "    return port\n",
    "\n",
    "\n",
    "async def write_config_and_run(\n",
    "    config: str, config_path: Union[str, Path], run_cmd: str\n",
    ") -> asyncio.subprocess.Process:\n",
    "    \"\"\"Writes the configuration to a file, and runs a command using the configuration.\n",
    "\n",
    "    Args:\n",
    "        config: The configuration string.\n",
    "        config_path: Path to the configuration file.\n",
    "        run_cmd: The command to run.\n",
    "\n",
    "    Returns:\n",
    "        The subprocess process object.\n",
    "    \"\"\"\n",
    "    with open(config_path, \"w\") as f:\n",
    "        f.write(config)\n",
    "\n",
    "    return await asyncio.create_subprocess_exec(\n",
    "        run_cmd,\n",
    "        config_path,\n",
    "        stdout=asyncio.subprocess.PIPE,\n",
    "        stdin=asyncio.subprocess.PIPE,\n",
    "    )\n",
    "\n",
    "\n",
    "@patch\n",
    "def get_service_config_string(\n",
    "    self: ApacheKafkaBroker, service: str, *, data_dir: Path\n",
    ") -> str:\n",
    "    \"\"\"Gets the configuration string for a service.\n",
    "\n",
    "    Args:\n",
    "        service: Name of the service (\"kafka\" or \"zookeeper\").\n",
    "        data_dir: Path to the directory where the service will save data.\n",
    "\n",
    "    Returns:\n",
    "        The service configuration string.\n",
    "    \"\"\"\n",
    "    service_kwargs = getattr(self, f\"{service}_kwargs\")\n",
    "    if service == \"kafka\":\n",
    "        return get_kafka_config_string(data_dir=data_dir, **service_kwargs)\n",
    "    else:\n",
    "        return get_zookeeper_config_string(data_dir=data_dir, **service_kwargs)\n",
    "\n",
    "\n",
    "@patch\n",
    "async def _start_service(self: ApacheKafkaBroker, service: str = \"kafka\") -> None:\n",
    "    \"\"\"Starts a service (kafka or zookeeper) asynchronously.\n",
    "\n",
    "    Args:\n",
    "        service: Name of the service (\"kafka\" or \"zookeeper\").\n",
    "    \"\"\"\n",
    "    logger.info(f\"Starting {service}...\")\n",
    "\n",
    "    if self.temporary_directory_path is None:\n",
    "        raise ValueError(\n",
    "            \"ApacheKafkaBroker._start_service(): self.temporary_directory_path is None, did you initialise it?\"\n",
    "        )\n",
    "\n",
    "    configs_tried: List[Dict[str, Any]] = []\n",
    "\n",
    "    for i in range(self.retries + 1):\n",
    "        configs_tried = configs_tried + [getattr(self, f\"{service}_kwargs\").copy()]\n",
    "\n",
    "        service_config_path = self.temporary_directory_path / f\"{service}.properties\"\n",
    "\n",
    "        with open(service_config_path, \"w\") as f:\n",
    "            f.write(\n",
    "                self.get_service_config_string(\n",
    "                    service, data_dir=self.temporary_directory_path\n",
    "                )\n",
    "            )\n",
    "\n",
    "        try:\n",
    "            port = (\n",
    "                self.zookeeper_kwargs[\"zookeeper_port\"]\n",
    "                if service == \"zookeeper\"\n",
    "                else self.kafka_kwargs[\"listener_port\"]\n",
    "            )\n",
    "            if is_port_in_use(port):\n",
    "                raise ValueError(f\"Port {port} is already in use\")\n",
    "\n",
    "            script_extension = \"bat\" if platform.system() == \"Windows\" else \"sh\"\n",
    "            service_start_script = f\"{service}-server-start.{script_extension}\"\n",
    "            service_task = await run_and_match(\n",
    "                service_start_script,\n",
    "                str(service_config_path),\n",
    "                pattern=\"Recorded new controller, from now on will use node\"\n",
    "                if service == \"kafka\"\n",
    "                else \"INFO Snapshot taken\",\n",
    "                timeout=30,\n",
    "            )\n",
    "        except Exception as e:\n",
    "            print(e)\n",
    "            logger.info(\n",
    "                f\"{service} startup failed, generating a new port and retrying...\"\n",
    "            )\n",
    "            port = get_free_port()\n",
    "            if service == \"zookeeper\":\n",
    "                self.zookeeper_kwargs[\"zookeeper_port\"] = port\n",
    "                self.kafka_kwargs[\"zookeeper_port\"] = port\n",
    "            else:\n",
    "                self.kafka_kwargs[\"listener_port\"] = port\n",
    "\n",
    "            logger.info(f\"{service} new port={port}\")\n",
    "        else:\n",
    "            setattr(self, f\"{service}_task\", service_task)\n",
    "            return\n",
    "\n",
    "    raise ValueError(f\"Could not start {service} with params: {configs_tried}\")\n",
    "\n",
    "\n",
    "@patch\n",
    "async def _start_kafka(self: ApacheKafkaBroker) -> None:\n",
    "    \"\"\"Starts a local Kafka broker asynchronously.\"\"\"\n",
    "    return await self._start_service(\"kafka\")\n",
    "\n",
    "\n",
    "@patch\n",
    "async def _start_zookeeper(self: ApacheKafkaBroker) -> None:\n",
    "    \"\"\"Starts a local ZooKeeper instance asynchronously.\"\"\"\n",
    "    return await self._start_service(\"zookeeper\")\n",
    "\n",
    "\n",
    "@patch\n",
    "async def _create_topics(self: ApacheKafkaBroker) -> None:\n",
    "    \"\"\"Creates missing topics in a local Kafka broker asynchronously.\"\"\"\n",
    "    listener_port = self.kafka_kwargs.get(\"listener_port\", 9092)\n",
    "    bootstrap_server = f\"127.0.0.1:{listener_port}\"\n",
    "\n",
    "    script_extension = \"bat\" if platform.system() == \"Windows\" else \"sh\"\n",
    "    topics_script = f\"kafka-topics.{script_extension}\"\n",
    "    async with asyncer.create_task_group() as tg:\n",
    "        processes = [\n",
    "            tg.soonify(asyncio.create_subprocess_exec)(\n",
    "                topics_script,\n",
    "                \"--create\",\n",
    "                f\"--topic={topic}\",\n",
    "                f\"--bootstrap-server={bootstrap_server}\",\n",
    "                stdout=asyncio.subprocess.PIPE,\n",
    "                stdin=asyncio.subprocess.PIPE,\n",
    "            )\n",
    "            for topic in self.topics\n",
    "        ]\n",
    "\n",
    "    try:\n",
    "        return_values = [\n",
    "            await asyncio.wait_for(process.value.wait(), 30) for process in processes\n",
    "        ]\n",
    "        if any(return_value != 0 for return_value in return_values):\n",
    "            raise ValueError(\"Could not create missing topics!\")\n",
    "    except asyncio.TimeoutError as _:\n",
    "        raise ValueError(\"Timed out while creating missing topics!\")\n",
    "\n",
    "\n",
    "@patch\n",
    "async def _start(self: ApacheKafkaBroker) -> str:\n",
    "    \"\"\"Starts a local Kafka broker and ZooKeeper instance asynchronously.\n",
    "\n",
    "    Returns:\n",
    "        The Kafka broker bootstrap server address in string format: host:port.\n",
    "    \"\"\"\n",
    "    self._check_deps()\n",
    "\n",
    "    self.temporary_directory = TemporaryDirectory()\n",
    "    self.temporary_directory_path = Path(self.temporary_directory.__enter__())\n",
    "\n",
    "    await self._start_zookeeper()\n",
    "    await self._start_kafka()\n",
    "\n",
    "    listener_port = self.kafka_kwargs.get(\"listener_port\", 9092)\n",
    "    bootstrap_server = f\"127.0.0.1:{listener_port}\"\n",
    "    logger.info(f\"Local Kafka broker up and running on {bootstrap_server}\")\n",
    "\n",
    "    await self._create_topics()\n",
    "\n",
    "    self._is_started = True\n",
    "\n",
    "    return bootstrap_server\n",
    "\n",
    "\n",
    "@patch\n",
    "async def _stop(self: ApacheKafkaBroker) -> None:\n",
    "    \"\"\"Stops a local Kafka broker and ZooKeeper instance asynchronously.\"\"\"\n",
    "    await terminate_asyncio_process(self.kafka_task)  # type: ignore\n",
    "    await terminate_asyncio_process(self.zookeeper_task)  # type: ignore\n",
    "    self.temporary_directory.__exit__(None, None, None)  # type: ignore\n",
    "    self._is_started = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "febe12c9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 47147...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 47147 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 46766...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 46766 terminated.\n",
      "**************************************************ZOOKEEPER LOGS++++++++++++++++++++++++++++++++++++++++++++++++++\n",
      "[2023-06-23 12:28:08,810] INFO PrepRequestProcessor (sid:0) started, reconfigEnabled=false (org.apache.zookeeper.server.PrepRequestProcessor)\n",
      "[2023-06-23 12:28:08,810] INFO zookeeper.request_throttler.shutdownTimeout = 10000 (org.apache.zookeeper.server.RequestThrottler)\n",
      "[2023-06-23 12:28:08,821] INFO Using checkIntervalMs=60000 maxPerMinute=10000 maxNeverUsedIntervalMs=0 (org.apache.zookeeper.server.ContainerManager)\n",
      "[2023-06-23 12:28:08,822] INFO ZooKeeper audit is disabled. (org.apache.zookeeper.audit.ZKAuditProvider)\n",
      "[2023-06-23 12:28:09,626] INFO Creating new log file: log.1 (org.apache.zookeeper.server.persistence.FileTxnLog)\n",
      "\n",
      "**************************************************KAFKA LOGS++++++++++++++++++++++++++++++++++++++++++++++++++\n",
      "[2023-06-23 12:28:10,617] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Recorded new controller, from now on will use node kumaran-fastkafka-devel:29092 (id: 0 rack: null) (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:11,583] INFO Terminating process due to signal SIGTERM (org.apache.kafka.common.utils.LoggingSignalHandler)\n",
      "[2023-06-23 12:28:11,584] INFO [KafkaServer id=0] shutting down (kafka.server.KafkaServer)\n",
      "[2023-06-23 12:28:11,585] INFO [KafkaServer id=0] Starting controlled shutdown (kafka.server.KafkaServer)\n",
      "[2023-06-23 12:28:11,596] INFO [KafkaServer id=0] Controlled shutdown request returned successfully after 7ms (kafka.server.KafkaServer)\n",
      "[2023-06-23 12:28:11,598] INFO [/config/changes-event-process-thread]: Shutting down (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)\n",
      "[2023-06-23 12:28:11,599] INFO [/config/changes-event-process-thread]: Stopped (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)\n",
      "[2023-06-23 12:28:11,599] INFO [/config/changes-event-process-thread]: Shutdown completed (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)\n",
      "[2023-06-23 12:28:11,599] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Stopping socket server request processors (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:11,603] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Stopped socket server request processors (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:11,604] INFO [data-plane Kafka Request Handler on Broker 0], shutting down (kafka.server.KafkaRequestHandlerPool)\n",
      "[2023-06-23 12:28:11,605] INFO [data-plane Kafka Request Handler on Broker 0], shut down completely (kafka.server.KafkaRequestHandlerPool)\n",
      "[2023-06-23 12:28:11,607] INFO [ExpirationReaper-0-AlterAcls]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,608] INFO [ExpirationReaper-0-AlterAcls]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,608] INFO [ExpirationReaper-0-AlterAcls]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,609] INFO [KafkaApi-0] Shutdown complete. (kafka.server.KafkaApis)\n",
      "[2023-06-23 12:28:11,609] INFO [ExpirationReaper-0-topic]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,610] INFO [ExpirationReaper-0-topic]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,610] INFO [ExpirationReaper-0-topic]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,611] INFO [TransactionCoordinator id=0] Shutting down. (kafka.coordinator.transaction.TransactionCoordinator)\n",
      "[2023-06-23 12:28:11,612] INFO [Transaction State Manager 0]: Shutdown complete (kafka.coordinator.transaction.TransactionStateManager)\n",
      "[2023-06-23 12:28:11,612] INFO [TxnMarkerSenderThread-0]: Shutting down (kafka.coordinator.transaction.TransactionMarkerChannelManager)\n",
      "[2023-06-23 12:28:11,612] INFO [TxnMarkerSenderThread-0]: Stopped (kafka.coordinator.transaction.TransactionMarkerChannelManager)\n",
      "[2023-06-23 12:28:11,612] INFO [TxnMarkerSenderThread-0]: Shutdown completed (kafka.coordinator.transaction.TransactionMarkerChannelManager)\n",
      "[2023-06-23 12:28:11,613] INFO [TransactionCoordinator id=0] Shutdown complete. (kafka.coordinator.transaction.TransactionCoordinator)\n",
      "[2023-06-23 12:28:11,613] INFO [GroupCoordinator 0]: Shutting down. (kafka.coordinator.group.GroupCoordinator)\n",
      "[2023-06-23 12:28:11,614] INFO [ExpirationReaper-0-Heartbeat]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,614] INFO [ExpirationReaper-0-Heartbeat]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,615] INFO [ExpirationReaper-0-Heartbeat]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,615] INFO [ExpirationReaper-0-Rebalance]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,615] INFO [ExpirationReaper-0-Rebalance]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,615] INFO [ExpirationReaper-0-Rebalance]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,615] INFO [GroupCoordinator 0]: Shutdown complete. (kafka.coordinator.group.GroupCoordinator)\n",
      "[2023-06-23 12:28:11,616] INFO [ReplicaManager broker=0] Shutting down (kafka.server.ReplicaManager)\n",
      "[2023-06-23 12:28:11,616] INFO [LogDirFailureHandler]: Shutting down (kafka.server.ReplicaManager$LogDirFailureHandler)\n",
      "[2023-06-23 12:28:11,616] INFO [LogDirFailureHandler]: Stopped (kafka.server.ReplicaManager$LogDirFailureHandler)\n",
      "[2023-06-23 12:28:11,616] INFO [LogDirFailureHandler]: Shutdown completed (kafka.server.ReplicaManager$LogDirFailureHandler)\n",
      "[2023-06-23 12:28:11,616] INFO [ReplicaFetcherManager on broker 0] shutting down (kafka.server.ReplicaFetcherManager)\n",
      "[2023-06-23 12:28:11,617] INFO [ReplicaFetcherManager on broker 0] shutdown completed (kafka.server.ReplicaFetcherManager)\n",
      "[2023-06-23 12:28:11,617] INFO [ReplicaAlterLogDirsManager on broker 0] shutting down (kafka.server.ReplicaAlterLogDirsManager)\n",
      "[2023-06-23 12:28:11,618] INFO [ReplicaAlterLogDirsManager on broker 0] shutdown completed (kafka.server.ReplicaAlterLogDirsManager)\n",
      "[2023-06-23 12:28:11,618] INFO [ExpirationReaper-0-Fetch]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,618] INFO [ExpirationReaper-0-Fetch]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,618] INFO [ExpirationReaper-0-Fetch]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,618] INFO [ExpirationReaper-0-Produce]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,619] INFO [ExpirationReaper-0-Produce]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,619] INFO [ExpirationReaper-0-Produce]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,619] INFO [ExpirationReaper-0-DeleteRecords]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,619] INFO [ExpirationReaper-0-DeleteRecords]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,619] INFO [ExpirationReaper-0-DeleteRecords]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,619] INFO [ExpirationReaper-0-ElectLeader]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,620] INFO [ExpirationReaper-0-ElectLeader]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,620] INFO [ExpirationReaper-0-ElectLeader]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:11,624] INFO [ReplicaManager broker=0] Shut down completely (kafka.server.ReplicaManager)\n",
      "[2023-06-23 12:28:11,625] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Shutting down (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:11,625] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Stopped (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:11,625] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Shutdown completed (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:11,627] INFO Broker to controller channel manager for alter-partition shutdown (kafka.server.BrokerToControllerChannelManagerImpl)\n",
      "[2023-06-23 12:28:11,627] INFO [zk-broker-0-to-controller-forwarding-channel-manager]: Shutting down (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:11,627] INFO [zk-broker-0-to-controller-forwarding-channel-manager]: Stopped (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:11,627] INFO [zk-broker-0-to-controller-forwarding-channel-manager]: Shutdown completed (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:11,627] INFO Broker to controller channel manager for forwarding shutdown (kafka.server.BrokerToControllerChannelManagerImpl)\n",
      "[2023-06-23 12:28:11,628] INFO Shutting down. (kafka.log.LogManager)\n",
      "[2023-06-23 12:28:11,629] INFO [kafka-log-cleaner-thread-0]: Shutting down (kafka.log.LogCleaner$CleanerThread)\n",
      "[2023-06-23 12:28:11,630] INFO [kafka-log-cleaner-thread-0]: Stopped (kafka.log.LogCleaner$CleanerThread)\n",
      "[2023-06-23 12:28:11,630] INFO [kafka-log-cleaner-thread-0]: Shutdown completed (kafka.log.LogCleaner$CleanerThread)\n",
      "[2023-06-23 12:28:11,641] INFO Shutdown complete. (kafka.log.LogManager)\n",
      "[2023-06-23 12:28:11,646] INFO [feature-zk-node-event-process-thread]: Shutting down (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)\n",
      "[2023-06-23 12:28:11,646] INFO [feature-zk-node-event-process-thread]: Shutdown completed (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)\n",
      "[2023-06-23 12:28:11,646] INFO [feature-zk-node-event-process-thread]: Stopped (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)\n",
      "[2023-06-23 12:28:11,646] INFO [ZooKeeperClient Kafka server] Closing. (kafka.zookeeper.ZooKeeperClient)\n",
      "[2023-06-23 12:28:11,750] INFO Session: 0x10001a4af720000 closed (org.apache.zookeeper.ZooKeeper)\n",
      "[2023-06-23 12:28:11,750] INFO EventThread shut down for session: 0x10001a4af720000 (org.apache.zookeeper.ClientCnxn)\n",
      "[2023-06-23 12:28:11,751] INFO [ZooKeeperClient Kafka server] Closed. (kafka.zookeeper.ZooKeeperClient)\n",
      "[2023-06-23 12:28:11,751] INFO [ThrottledChannelReaper-Fetch]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,752] INFO [ThrottledChannelReaper-Fetch]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,752] INFO [ThrottledChannelReaper-Fetch]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,752] INFO [ThrottledChannelReaper-Produce]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,752] INFO [ThrottledChannelReaper-Produce]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,752] INFO [ThrottledChannelReaper-Produce]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,753] INFO [ThrottledChannelReaper-Request]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,753] INFO [ThrottledChannelReaper-Request]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,753] INFO [ThrottledChannelReaper-Request]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,753] INFO [ThrottledChannelReaper-ControllerMutation]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,753] INFO [ThrottledChannelReaper-ControllerMutation]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,753] INFO [ThrottledChannelReaper-ControllerMutation]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:11,754] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Shutting down socket server (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:11,763] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Shutdown completed (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:11,763] INFO Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics)\n",
      "[2023-06-23 12:28:11,763] INFO Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics)\n",
      "[2023-06-23 12:28:11,763] INFO Metrics reporters closed (org.apache.kafka.common.metrics.Metrics)\n",
      "[2023-06-23 12:28:11,764] INFO Broker and topic stats closed (kafka.server.BrokerTopicStats)\n",
      "[2023-06-23 12:28:11,764] INFO App info kafka.server for 0 unregistered (org.apache.kafka.common.utils.AppInfoParser)\n",
      "[2023-06-23 12:28:11,765] INFO [KafkaServer id=0] shut down completed (kafka.server.KafkaServer)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "broker = ApacheKafkaBroker(listener_port=29092)\n",
    "async with broker:\n",
    "    pass\n",
    "\n",
    "print(\"*\" * 50 + \"ZOOKEEPER LOGS\" + \"+\" * 50)\n",
    "zookeeper_output, _ = await broker.zookeeper_task.communicate()\n",
    "print(zookeeper_output.decode(\"UTF-8\"))\n",
    "\n",
    "print(\"*\" * 50 + \"KAFKA LOGS\" + \"+\" * 50)\n",
    "kafka_output, _ = await broker.kafka_task.communicate()\n",
    "print(kafka_output.decode(\"UTF-8\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "10338a30",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "Port 2181 is already in use\n",
      "[INFO] __main__: zookeeper startup failed, generating a new port and retrying...\n",
      "[INFO] __main__: zookeeper new port=34331\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 48014...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 48014 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 47632...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 47632 terminated.\n"
     ]
    }
   ],
   "source": [
    "broker_1 = ApacheKafkaBroker(listener_port=29092)\n",
    "async with broker_1:\n",
    "    port = broker_1.zookeeper_kwargs[\"zookeeper_port\"]\n",
    "    broker_2 = ApacheKafkaBroker(zookeeper_port=port, retries=0)\n",
    "    with pytest.raises(ValueError) as e:\n",
    "        async with broker_2:\n",
    "            pass\n",
    "\n",
    "assert e.value.args[0].startswith(\"Could not start zookeeper with params:\")\n",
    "\n",
    "for broker in [broker_2]:\n",
    "    assert broker.zookeeper_task == None\n",
    "#     print(\"*\" * 50 + \"ZOOKEEPER LOGS\" + \"+\" * 50)\n",
    "#     zookeeper_output, _ = await broker.zookeeper_task.communicate()\n",
    "#     print(zookeeper_output.decode(\"UTF-8\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c0aff342",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "@patch\n",
    "def start(self: ApacheKafkaBroker) -> str:\n",
    "    \"\"\"Starts a local Kafka broker and ZooKeeper instance synchronously.\n",
    "\n",
    "    Returns:\n",
    "        The Kafka broker bootstrap server address in string format: host:port.\n",
    "    \"\"\"\n",
    "    logger.info(f\"{self.__class__.__name__}.start(): entering...\")\n",
    "    try:\n",
    "        # get or create loop\n",
    "        try:\n",
    "            loop = asyncio.get_event_loop()\n",
    "        except RuntimeError as e:\n",
    "            logger.warning(\n",
    "                f\"{self.__class__.__name__}.start(): RuntimeError raised when calling asyncio.get_event_loop(): {e}\"\n",
    "            )\n",
    "            logger.warning(\n",
    "                f\"{self.__class__.__name__}.start(): asyncio.new_event_loop()\"\n",
    "            )\n",
    "            loop = asyncio.new_event_loop()\n",
    "\n",
    "        # start zookeeper and kafka broker in the loop\n",
    "\n",
    "        if loop.is_running():\n",
    "            if self.apply_nest_asyncio:\n",
    "                logger.warning(\n",
    "                    f\"{self.__class__.__name__}.start(): ({loop}) is already running!\"\n",
    "                )\n",
    "                logger.warning(\n",
    "                    f\"{self.__class__.__name__}.start(): calling nest_asyncio.apply()\"\n",
    "                )\n",
    "                nest_asyncio.apply(loop)\n",
    "            else:\n",
    "                msg = f\"{self.__class__.__name__}.start(): ({loop}) is already running! Use 'apply_nest_asyncio=True' when creating 'ApacheKafkaBroker' to prevent this.\"\n",
    "                logger.error(msg)\n",
    "                raise RuntimeError(msg)\n",
    "\n",
    "        retval = loop.run_until_complete(self._start())\n",
    "        logger.info(f\"{self.__class__}.start(): returning {retval}\")\n",
    "        return retval\n",
    "    finally:\n",
    "        logger.info(f\"{self.__class__.__name__}.start(): exited.\")\n",
    "\n",
    "\n",
    "@patch\n",
    "def stop(self: ApacheKafkaBroker) -> None:\n",
    "    \"\"\"Stops a local kafka broker and zookeeper instance synchronously\"\"\"\n",
    "    logger.info(f\"{self.__class__.__name__}.stop(): entering...\")\n",
    "    try:\n",
    "        if not self._is_started:\n",
    "            raise RuntimeError(\n",
    "                \"ApacheKafkaBroker not started yet, please call ApacheKafkaBroker.start() before!\"\n",
    "            )\n",
    "\n",
    "        loop = asyncio.get_event_loop()\n",
    "        loop.run_until_complete(self._stop())\n",
    "    finally:\n",
    "        logger.info(f\"{self.__class__.__name__}.stop(): exited.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "23db243b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] __main__: ApacheKafkaBroker.start(): entering...\n",
      "[WARNING] __main__: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "[WARNING] __main__: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "[INFO] __main__: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:29092\n",
      "[INFO] __main__: ApacheKafkaBroker.start(): exited.\n",
      "Hello world!\n",
      "[INFO] __main__: ApacheKafkaBroker.stop(): entering...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 48880...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 48880 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 48500...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 48500 terminated.\n",
      "[INFO] __main__: ApacheKafkaBroker.stop(): exited.\n",
      "**************************************************ZOOKEEPER LOGS++++++++++++++++++++++++++++++++++++++++++++++++++\n",
      "[2023-06-23 12:28:19,686] INFO PrepRequestProcessor (sid:0) started, reconfigEnabled=false (org.apache.zookeeper.server.PrepRequestProcessor)\n",
      "[2023-06-23 12:28:19,687] INFO zookeeper.request_throttler.shutdownTimeout = 10000 (org.apache.zookeeper.server.RequestThrottler)\n",
      "[2023-06-23 12:28:19,699] INFO Using checkIntervalMs=60000 maxPerMinute=10000 maxNeverUsedIntervalMs=0 (org.apache.zookeeper.server.ContainerManager)\n",
      "[2023-06-23 12:28:19,700] INFO ZooKeeper audit is disabled. (org.apache.zookeeper.audit.ZKAuditProvider)\n",
      "[2023-06-23 12:28:20,632] INFO Creating new log file: log.1 (org.apache.zookeeper.server.persistence.FileTxnLog)\n",
      "\n",
      "**************************************************KAFKA LOGS++++++++++++++++++++++++++++++++++++++++++++++++++\n",
      "[2023-06-23 12:28:21,706] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Recorded new controller, from now on will use node kumaran-fastkafka-devel:29092 (id: 0 rack: null) (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:22,660] INFO Terminating process due to signal SIGTERM (org.apache.kafka.common.utils.LoggingSignalHandler)\n",
      "[2023-06-23 12:28:22,661] INFO [KafkaServer id=0] shutting down (kafka.server.KafkaServer)\n",
      "[2023-06-23 12:28:22,662] INFO [KafkaServer id=0] Starting controlled shutdown (kafka.server.KafkaServer)\n",
      "[2023-06-23 12:28:22,673] INFO [KafkaServer id=0] Controlled shutdown request returned successfully after 7ms (kafka.server.KafkaServer)\n",
      "[2023-06-23 12:28:22,676] INFO [/config/changes-event-process-thread]: Shutting down (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)\n",
      "[2023-06-23 12:28:22,676] INFO [/config/changes-event-process-thread]: Stopped (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)\n",
      "[2023-06-23 12:28:22,676] INFO [/config/changes-event-process-thread]: Shutdown completed (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)\n",
      "[2023-06-23 12:28:22,677] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Stopping socket server request processors (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:22,682] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Stopped socket server request processors (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:22,682] INFO [data-plane Kafka Request Handler on Broker 0], shutting down (kafka.server.KafkaRequestHandlerPool)\n",
      "[2023-06-23 12:28:22,683] INFO [data-plane Kafka Request Handler on Broker 0], shut down completely (kafka.server.KafkaRequestHandlerPool)\n",
      "[2023-06-23 12:28:22,685] INFO [ExpirationReaper-0-AlterAcls]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,686] INFO [ExpirationReaper-0-AlterAcls]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,686] INFO [ExpirationReaper-0-AlterAcls]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,687] INFO [KafkaApi-0] Shutdown complete. (kafka.server.KafkaApis)\n",
      "[2023-06-23 12:28:22,687] INFO [ExpirationReaper-0-topic]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,687] INFO [ExpirationReaper-0-topic]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,687] INFO [ExpirationReaper-0-topic]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,689] INFO [TransactionCoordinator id=0] Shutting down. (kafka.coordinator.transaction.TransactionCoordinator)\n",
      "[2023-06-23 12:28:22,689] INFO [Transaction State Manager 0]: Shutdown complete (kafka.coordinator.transaction.TransactionStateManager)\n",
      "[2023-06-23 12:28:22,689] INFO [TxnMarkerSenderThread-0]: Shutting down (kafka.coordinator.transaction.TransactionMarkerChannelManager)\n",
      "[2023-06-23 12:28:22,690] INFO [TxnMarkerSenderThread-0]: Stopped (kafka.coordinator.transaction.TransactionMarkerChannelManager)\n",
      "[2023-06-23 12:28:22,690] INFO [TxnMarkerSenderThread-0]: Shutdown completed (kafka.coordinator.transaction.TransactionMarkerChannelManager)\n",
      "[2023-06-23 12:28:22,690] INFO [TransactionCoordinator id=0] Shutdown complete. (kafka.coordinator.transaction.TransactionCoordinator)\n",
      "[2023-06-23 12:28:22,690] INFO [GroupCoordinator 0]: Shutting down. (kafka.coordinator.group.GroupCoordinator)\n",
      "[2023-06-23 12:28:22,691] INFO [ExpirationReaper-0-Heartbeat]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,691] INFO [ExpirationReaper-0-Heartbeat]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,691] INFO [ExpirationReaper-0-Heartbeat]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,692] INFO [ExpirationReaper-0-Rebalance]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,692] INFO [ExpirationReaper-0-Rebalance]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,692] INFO [ExpirationReaper-0-Rebalance]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,692] INFO [GroupCoordinator 0]: Shutdown complete. (kafka.coordinator.group.GroupCoordinator)\n",
      "[2023-06-23 12:28:22,693] INFO [ReplicaManager broker=0] Shutting down (kafka.server.ReplicaManager)\n",
      "[2023-06-23 12:28:22,693] INFO [LogDirFailureHandler]: Shutting down (kafka.server.ReplicaManager$LogDirFailureHandler)\n",
      "[2023-06-23 12:28:22,693] INFO [LogDirFailureHandler]: Stopped (kafka.server.ReplicaManager$LogDirFailureHandler)\n",
      "[2023-06-23 12:28:22,693] INFO [LogDirFailureHandler]: Shutdown completed (kafka.server.ReplicaManager$LogDirFailureHandler)\n",
      "[2023-06-23 12:28:22,694] INFO [ReplicaFetcherManager on broker 0] shutting down (kafka.server.ReplicaFetcherManager)\n",
      "[2023-06-23 12:28:22,694] INFO [ReplicaFetcherManager on broker 0] shutdown completed (kafka.server.ReplicaFetcherManager)\n",
      "[2023-06-23 12:28:22,695] INFO [ReplicaAlterLogDirsManager on broker 0] shutting down (kafka.server.ReplicaAlterLogDirsManager)\n",
      "[2023-06-23 12:28:22,695] INFO [ReplicaAlterLogDirsManager on broker 0] shutdown completed (kafka.server.ReplicaAlterLogDirsManager)\n",
      "[2023-06-23 12:28:22,695] INFO [ExpirationReaper-0-Fetch]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,695] INFO [ExpirationReaper-0-Fetch]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,695] INFO [ExpirationReaper-0-Fetch]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,696] INFO [ExpirationReaper-0-Produce]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,696] INFO [ExpirationReaper-0-Produce]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,696] INFO [ExpirationReaper-0-Produce]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,696] INFO [ExpirationReaper-0-DeleteRecords]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,696] INFO [ExpirationReaper-0-DeleteRecords]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,696] INFO [ExpirationReaper-0-DeleteRecords]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,697] INFO [ExpirationReaper-0-ElectLeader]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,697] INFO [ExpirationReaper-0-ElectLeader]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,697] INFO [ExpirationReaper-0-ElectLeader]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)\n",
      "[2023-06-23 12:28:22,702] INFO [ReplicaManager broker=0] Shut down completely (kafka.server.ReplicaManager)\n",
      "[2023-06-23 12:28:22,702] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Shutting down (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:22,702] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Stopped (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:22,702] INFO [zk-broker-0-to-controller-alter-partition-channel-manager]: Shutdown completed (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:22,703] INFO Broker to controller channel manager for alter-partition shutdown (kafka.server.BrokerToControllerChannelManagerImpl)\n",
      "[2023-06-23 12:28:22,704] INFO [zk-broker-0-to-controller-forwarding-channel-manager]: Shutting down (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:22,704] INFO [zk-broker-0-to-controller-forwarding-channel-manager]: Stopped (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:22,704] INFO [zk-broker-0-to-controller-forwarding-channel-manager]: Shutdown completed (kafka.server.BrokerToControllerRequestThread)\n",
      "[2023-06-23 12:28:22,704] INFO Broker to controller channel manager for forwarding shutdown (kafka.server.BrokerToControllerChannelManagerImpl)\n",
      "[2023-06-23 12:28:22,704] INFO Shutting down. (kafka.log.LogManager)\n",
      "[2023-06-23 12:28:22,705] INFO [kafka-log-cleaner-thread-0]: Shutting down (kafka.log.LogCleaner$CleanerThread)\n",
      "[2023-06-23 12:28:22,706] INFO [kafka-log-cleaner-thread-0]: Shutdown completed (kafka.log.LogCleaner$CleanerThread)\n",
      "[2023-06-23 12:28:22,706] INFO [kafka-log-cleaner-thread-0]: Stopped (kafka.log.LogCleaner$CleanerThread)\n",
      "[2023-06-23 12:28:22,724] INFO Shutdown complete. (kafka.log.LogManager)\n",
      "[2023-06-23 12:28:22,729] INFO [feature-zk-node-event-process-thread]: Shutting down (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)\n",
      "[2023-06-23 12:28:22,729] INFO [feature-zk-node-event-process-thread]: Shutdown completed (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)\n",
      "[2023-06-23 12:28:22,729] INFO [feature-zk-node-event-process-thread]: Stopped (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread)\n",
      "[2023-06-23 12:28:22,729] INFO [ZooKeeperClient Kafka server] Closing. (kafka.zookeeper.ZooKeeperClient)\n",
      "[2023-06-23 12:28:22,833] INFO Session: 0x10001a4d9ec0000 closed (org.apache.zookeeper.ZooKeeper)\n",
      "[2023-06-23 12:28:22,833] INFO EventThread shut down for session: 0x10001a4d9ec0000 (org.apache.zookeeper.ClientCnxn)\n",
      "[2023-06-23 12:28:22,834] INFO [ZooKeeperClient Kafka server] Closed. (kafka.zookeeper.ZooKeeperClient)\n",
      "[2023-06-23 12:28:22,834] INFO [ThrottledChannelReaper-Fetch]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,835] INFO [ThrottledChannelReaper-Fetch]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,835] INFO [ThrottledChannelReaper-Fetch]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,835] INFO [ThrottledChannelReaper-Produce]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-Produce]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-Produce]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-Request]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-Request]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-Request]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-ControllerMutation]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-ControllerMutation]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,836] INFO [ThrottledChannelReaper-ControllerMutation]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)\n",
      "[2023-06-23 12:28:22,837] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Shutting down socket server (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:22,846] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Shutdown completed (kafka.network.SocketServer)\n",
      "[2023-06-23 12:28:22,846] INFO Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics)\n",
      "[2023-06-23 12:28:22,846] INFO Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics)\n",
      "[2023-06-23 12:28:22,846] INFO Metrics reporters closed (org.apache.kafka.common.metrics.Metrics)\n",
      "[2023-06-23 12:28:22,847] INFO Broker and topic stats closed (kafka.server.BrokerTopicStats)\n",
      "[2023-06-23 12:28:22,848] INFO App info kafka.server for 0 unregistered (org.apache.kafka.common.utils.AppInfoParser)\n",
      "[2023-06-23 12:28:22,848] INFO [KafkaServer id=0] shut down completed (kafka.server.KafkaServer)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "broker = ApacheKafkaBroker(apply_nest_asyncio=True, listener_port=29092)\n",
    "with broker:\n",
    "    print(\"Hello world!\")\n",
    "\n",
    "print(\"*\" * 50 + \"ZOOKEEPER LOGS\" + \"+\" * 50)\n",
    "zookeeper_output, _ = await broker.zookeeper_task.communicate()\n",
    "print(zookeeper_output.decode(\"UTF-8\"))\n",
    "\n",
    "\n",
    "print(\"*\" * 50 + \"KAFKA LOGS\" + \"+\" * 50)\n",
    "kafka_output, _ = await broker.kafka_task.communicate()\n",
    "print(kafka_output.decode(\"UTF-8\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "17c5fa47",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 49747...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 49747 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 49365...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 49365 terminated.\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "topics = [\"topic_1\", \"topic_2\"]\n",
    "script_extension = \"bat\" if platform.system() == \"Windows\" else \"sh\"\n",
    "topics_script = f\"kafka-topics.{script_extension}\"\n",
    "\n",
    "async with ApacheKafkaBroker(topics=topics, listener_port=29092) as bootstrap_server:\n",
    "    task = await asyncio.create_subprocess_exec(\n",
    "        topics_script,\n",
    "        \"--list\",\n",
    "        f\"--bootstrap-server={bootstrap_server}\",\n",
    "        stdout=asyncio.subprocess.PIPE,\n",
    "        stdin=asyncio.subprocess.PIPE,\n",
    "    )\n",
    "    output, _ = await asyncio.wait_for(task.communicate(), 30)\n",
    "    listed_topics = output.decode(\"UTF-8\").split(\"\\n\")[:-1]\n",
    "    for i, topic in enumerate(listed_topics):\n",
    "        listed_topics[i] = topic.strip()\n",
    "    assert set(listed_topics) == set(topics), set(listed_topics)\n",
    "print(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1cc7e1a9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] __main__: ApacheKafkaBroker.start(): entering...\n",
      "[WARNING] __main__: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "[WARNING] __main__: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "[INFO] __main__: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:29092\n",
      "[INFO] __main__: ApacheKafkaBroker.start(): exited.\n",
      "[INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'test-topic'})\n",
      "[INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'test-topic': 1}. \n",
      "[INFO] __main__: ApacheKafkaBroker.stop(): entering...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 51689...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 51689 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 51307...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 51307 terminated.\n",
      "[INFO] __main__: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "test_topic = \"test-topic\"\n",
    "test_msg = b\"test-msg\"\n",
    "\n",
    "with ApacheKafkaBroker(\n",
    "    topics=[test_topic], apply_nest_asyncio=True, listener_port=29092\n",
    ") as bootstrap_server:\n",
    "    consumer = AIOKafkaConsumer(test_topic, bootstrap_servers=bootstrap_server)\n",
    "\n",
    "    producer = AIOKafkaProducer(bootstrap_servers=bootstrap_server)\n",
    "\n",
    "    await consumer.start()\n",
    "    await producer.start()\n",
    "\n",
    "    try:\n",
    "        await producer.send_and_wait(test_topic, test_msg)\n",
    "        msg = await consumer.getone()\n",
    "        assert msg, value == test_msg\n",
    "    finally:\n",
    "        await consumer.stop()\n",
    "        await producer.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "225d7ef3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "[INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'test-topic'})\n",
      "[INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'test-topic': 1}. \n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 52918...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 52918 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 52535...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 52535 terminated.\n"
     ]
    }
   ],
   "source": [
    "test_topic = \"test-topic\"\n",
    "test_msg = b\"test-msg\"\n",
    "\n",
    "async with ApacheKafkaBroker(\n",
    "    topics=[test_topic], listener_port=29092\n",
    ") as bootstrap_server:\n",
    "    consumer = AIOKafkaConsumer(test_topic, bootstrap_servers=bootstrap_server)\n",
    "\n",
    "    producer = AIOKafkaProducer(bootstrap_servers=bootstrap_server)\n",
    "\n",
    "    await consumer.start()\n",
    "    await producer.start()\n",
    "\n",
    "    try:\n",
    "        await producer.send_and_wait(test_topic, test_msg)\n",
    "        msg = await consumer.getone()\n",
    "        assert msg, value == test_msg\n",
    "    finally:\n",
    "        await consumer.stop()\n",
    "        await producer.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9b81e6bf",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "async def _start_broker(broker: Any) -> Union[Any, Exception]:\n",
    "    try:\n",
    "        await broker._start()\n",
    "        return broker\n",
    "    except Exception as e:\n",
    "        return e\n",
    "\n",
    "\n",
    "async def _stop_broker(broker: Any) -> Union[Any, Exception]:\n",
    "    try:\n",
    "        await broker._stop()\n",
    "        return broker\n",
    "    except Exception as e:\n",
    "        return e\n",
    "\n",
    "\n",
    "async def _get_unique_local_brokers_to_start(\n",
    "    kafka_brokers_name: str,\n",
    "    kafka_brokers: List[Dict[str, Dict[str, Any]]],\n",
    "    duplicate_ok: bool = False,\n",
    "    zookeeper_ports: List[int] = [2181],\n",
    "    ignore_nonlocal_brokers: bool = False,\n",
    ") -> List[Tuple[str, int]]:\n",
    "    brokers_to_start = [\n",
    "        x[kafka_brokers_name] for x in kafka_brokers if kafka_brokers_name in x\n",
    "    ]\n",
    "    unique_brokers_to_start = set([(x[\"url\"], x[\"port\"]) for x in brokers_to_start])\n",
    "\n",
    "    if len(unique_brokers_to_start) < len(brokers_to_start) and not duplicate_ok:\n",
    "        raise ValueError(\n",
    "            f\"Duplicate kafka_brokers are found - {brokers_to_start}. Please change values or use 'duplicate_ok=True'\"\n",
    "        )\n",
    "\n",
    "    unique_urls = set([x[0] for x in unique_brokers_to_start])\n",
    "    localhost_urls = set(\n",
    "        [\n",
    "            \"localhost\",\n",
    "            \"127.0.0.1\",\n",
    "            \"0.0.0.0\",  # nosec: B104 - Possible binding to all interfaces\n",
    "        ]\n",
    "    )\n",
    "    if not unique_urls.issubset(localhost_urls) and not ignore_nonlocal_brokers:\n",
    "        raise ValueError(\n",
    "            f\"URL values other than {', '.join(sorted(localhost_urls))} are found - {unique_urls - localhost_urls}. Please change values or use 'ignore_nonlocal_brokers=True'\"\n",
    "        )\n",
    "\n",
    "    unique_local_brokers_to_start = [\n",
    "        x for x in unique_brokers_to_start if x[0] in localhost_urls\n",
    "    ]\n",
    "    return unique_local_brokers_to_start"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "037b1905",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "T = TypeVar(\"T\")\n",
    "\n",
    "\n",
    "@asynccontextmanager\n",
    "async def _start_and_stop_brokers(brokers: List[T]) -> AsyncIterator[None]:\n",
    "    try:\n",
    "        retvals = [await _start_broker(broker) for broker in brokers]\n",
    "        exceptions = [x for x in retvals if isinstance(x, Exception)]\n",
    "\n",
    "        if exceptions:\n",
    "            raise RuntimeError(exceptions)\n",
    "\n",
    "        yield\n",
    "    finally:\n",
    "        retvals = [\n",
    "            await _stop_broker(broker)\n",
    "            for broker in retvals\n",
    "            if not isinstance(broker, Exception)\n",
    "        ]\n",
    "        exceptions = [x for x in retvals if isinstance(x, Exception)]\n",
    "\n",
    "        if exceptions:\n",
    "            raise RuntimeError(exceptions)\n",
    "\n",
    "\n",
    "@asynccontextmanager\n",
    "async def start_apache_kafka_brokers(\n",
    "    kafka_brokers_name: str,\n",
    "    kafka_brokers: List[Dict[str, Dict[str, Any]]],\n",
    "    duplicate_ok: bool = False,\n",
    "    zookeeper_ports: List[int] = [2181],\n",
    "    ignore_nonlocal_brokers: bool = False,\n",
    ") -> AsyncIterator[None]:\n",
    "    unique_local_brokers_to_start = await _get_unique_local_brokers_to_start(\n",
    "        kafka_brokers_name=kafka_brokers_name,\n",
    "        kafka_brokers=kafka_brokers,\n",
    "        duplicate_ok=duplicate_ok,\n",
    "        ignore_nonlocal_brokers=ignore_nonlocal_brokers,\n",
    "    )\n",
    "\n",
    "    if len(zookeeper_ports) < len(unique_local_brokers_to_start):\n",
    "        raise ValueError(\n",
    "            f\"Atleast {len(unique_local_brokers_to_start)} zookeeper ports are needed to start kafka. Current zookeeper_ports length is {len(zookeeper_ports)}\"\n",
    "        )\n",
    "\n",
    "    brokers = [\n",
    "        ApacheKafkaBroker(listener_port=broker[1], zookeeper_port=zookeeper_port)  # type: ignore\n",
    "        for broker, zookeeper_port in zip(\n",
    "            unique_local_brokers_to_start, zookeeper_ports\n",
    "        )\n",
    "    ]\n",
    "\n",
    "    async with _start_and_stop_brokers(brokers=brokers):\n",
    "        yield"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "43b93840",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29093\n",
      "Succesfully started multiple brokers\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 54147...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 54147 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 53766...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 53766 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 54998...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 54998 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 54616...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 54616 terminated.\n"
     ]
    }
   ],
   "source": [
    "kafka_broker_1 = dict(\n",
    "    development=dict(url=\"localhost\", port=29092),\n",
    "    production=dict(url=\"prod.server_1\", port=9092),\n",
    ")\n",
    "\n",
    "kafka_broker_2 = dict(\n",
    "    development=dict(url=\"localhost\", port=29093),\n",
    "    production=dict(url=\"prod.server_1\", port=9092),\n",
    ")\n",
    "\n",
    "kafka_brokers = [kafka_broker_1, kafka_broker_2]\n",
    "\n",
    "async with start_apache_kafka_brokers(\n",
    "    kafka_brokers_name=\"development\",\n",
    "    kafka_brokers=kafka_brokers,\n",
    "    zookeeper_ports=[22181, 22182],\n",
    "):\n",
    "    print(\"Succesfully started multiple brokers\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "955e9828",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<ExceptionInfo ValueError(\"Duplicate kafka_brokers are found - [{'url': 'localhost', 'port': 29092}, {'url': 'localhost', 'port': 29092}]. Please change values or use 'duplicate_ok=True'\") tblen=4>\n"
     ]
    }
   ],
   "source": [
    "kafka_brokers = [\n",
    "    dict(\n",
    "        development=dict(url=\"localhost\", port=29092),\n",
    "    ),\n",
    "    dict(\n",
    "        development=dict(url=\"localhost\", port=29092),\n",
    "    ),\n",
    "]\n",
    "\n",
    "with pytest.raises(ValueError) as e:\n",
    "    async with start_apache_kafka_brokers(\n",
    "        kafka_brokers_name=\"development\",\n",
    "        kafka_brokers=kafka_brokers,\n",
    "        duplicate_ok=False,\n",
    "    ):\n",
    "        print(\"If this is printed then the test didn't pass\")\n",
    "print(e)\n",
    "assert \"Duplicate kafka_brokers are found\" in str(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f6ac313",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<ExceptionInfo ValueError(\"URL values other than 0.0.0.0, 127.0.0.1, localhost are found - {'some_other_host'}. Please change values or use 'ignore_nonlocal_brokers=True'\") tblen=4>\n",
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] __main__: Starting zookeeper...\n",
      "[INFO] __main__: Starting kafka...\n",
      "[INFO] __main__: Local Kafka broker up and running on 127.0.0.1:29092\n",
      "Starting again with 'ignore_nonlocal_brokers=True'\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 55882...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 55882 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 55501...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 55501 terminated.\n"
     ]
    }
   ],
   "source": [
    "kafka_brokers = [\n",
    "    dict(\n",
    "        development=dict(url=\"127.0.0.1\", port=29092),\n",
    "    ),\n",
    "    dict(\n",
    "        development=dict(url=\"some_other_host\", port=29092),\n",
    "    ),\n",
    "]\n",
    "\n",
    "with pytest.raises(ValueError) as e:\n",
    "    async with start_apache_kafka_brokers(\n",
    "        kafka_brokers_name=\"development\",\n",
    "        kafka_brokers=kafka_brokers,\n",
    "        ignore_nonlocal_brokers=False,\n",
    "    ):\n",
    "        print(\"If this is printed then the test didn't pass\")\n",
    "print(e)\n",
    "assert \"URL values other than 0.0.0.0, 127.0.0.1, localhost are found\" in str(e)\n",
    "\n",
    "async with start_apache_kafka_brokers(\n",
    "    kafka_brokers_name=\"development\",\n",
    "    kafka_brokers=kafka_brokers,\n",
    "    ignore_nonlocal_brokers=True,\n",
    "):\n",
    "    print(\"Starting again with 'ignore_nonlocal_brokers=True'\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "24d6fbe7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<ExceptionInfo ValueError('Atleast 2 zookeeper ports are needed to start kafka. Current zookeeper_ports length is 1') tblen=3>\n"
     ]
    }
   ],
   "source": [
    "kafka_brokers = [\n",
    "    dict(\n",
    "        development=dict(url=\"localhost\", port=29092),\n",
    "    ),\n",
    "    dict(\n",
    "        development=dict(url=\"localhost\", port=29093),\n",
    "    ),\n",
    "]\n",
    "\n",
    "with pytest.raises(ValueError) as e:\n",
    "    async with start_apache_kafka_brokers(\n",
    "        kafka_brokers_name=\"development\",\n",
    "        kafka_brokers=kafka_brokers,\n",
    "        zookeeper_ports=[22181],\n",
    "    ):\n",
    "        print(\"If this is printed then the test didn't pass\")\n",
    "print(e)\n",
    "assert (\n",
    "    \"Atleast 2 zookeeper ports are needed to start kafka. Current zookeeper_ports length is 1\"\n",
    "    in str(e)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f67fecd5",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
